package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: guoml
 * @Description: 针对企查查t_patent表和中数t_sipo_patentinfo,t_sipo_patent_lawstateinfo表进行数据映射开发
 * @Date: 2021/12/7 12:51
 */

public class TPatent {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_patent").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_sipo_patentinfo");
        long zs_count = zs_data.count();
        if(zs_count>0){

            //获取中数增量数据表中的更新数据（inner join）
            Dataset<Row> update_data = spark.sql("SELECT a.id,\n" +
                    "       a.key_no,\n" +
                    "       a.company_id,\n" +
                    "       d.company_name,\n" +
                    "       a.patent_id,\n" +
                    "       b.PTNAME,\n" +
                    "       b.REG_NO,\n" +
                    "       b.REGDATE,\n" +
                    "       b.PUBCODE,\n" +
                    "       b.PUBDATE,\n" +
                    "       b.INVENTOR,\n" +
                    "       b.REGPER,\n" +
                    "       b.AGENTORG,\n" +
                    "       b.AGENT,\n" +
                    "       b.TYPECODE,\n" +
                    "       a.ipc_desc,\n" +
                    "       if(b.TYPENO='发明专利','发明公布',b.TYPENO),\n" +
                    "       c.LAWSTATUS,\n" +
                    "       c.LAWSTATUSDATE,\n" +
                    "       a.primary_examiner,\n" +
                    "       a.assiant_examiner,\n" +
                    "       a.patent_image,\n" +
                    "       a.other_references,\n" +
                    "       b.SUMMARY,\n" +
                    "       a.patent_legal_history,\n" +
                    "       a.document_types,\n" +
                    "       a.cites,\n" +
                    "       b.JOBID,\n" +
                    "       \"0\" as isadd\n" +
                    "FROM\n" +
                    "  (SELECT *\n" +
                    "   FROM a_dc_ep_ods.zs_t_sipo_patentinfo\n" +
                    "   WHERE CREDITCODE IS NOT NULL\n" +
                    "     AND CREDITCODE !='') b\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.zs_t_sipo_patent_lawstateinfo c on b.REG_NO = c.REG_NO\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.t_eci_company d on upper(trim(d.credit_code)) = upper(trim(b.CREDITCODE))\n" +
                    "INNER JOIN\n" +
                    "  (SELECT m.*,n.credit_code\n" +
                    "   FROM a_dc_ep_dwi.t_patent m\n" +
                    "INNER JOIN\n" +
                    "   a_dc_ep_ods.t_eci_company n\n" +
                    "ON upper(trim(m.company_id)) = upper(trim(n.company_id))\n" +
                    "   WHERE n.credit_code IS NOT NULL\n " +
                    "     AND n.credit_code!='') a ON upper(trim(a.credit_code)) = upper(trim(b.CREDITCODE)) and a.application_number = b.REG_NO and a.publication_number = b.PUBCODE WHERE d.credit_code IS NOT NULL and d.credit_code!=''");

            update_data.createOrReplaceTempView("tmp_t_patent");
            spark.sql("insert overwrite table a_dc_ep_incr.t_patent select * from tmp_t_patent");
            //获取中数增量数据表中的新增数据（left join）
            Dataset<Row> insert_data = spark.sql("SELECT b.RECORD_ID,\n" +
                    "       d.key_no,\n" +
                    "       d.company_id,\n" +
                    "       d.company_name,\n" +
                    "       '' as patent_id,\n" +
                    "       b.PTNAME,\n" +
                    "       b.REG_NO,\n" +
                    "       b.REGDATE,\n" +
                    "       b.PUBCODE,\n" +
                    "       b.PUBDATE,\n" +
                    "       b.INVENTOR,\n" +
                    "       b.REGPER,\n" +
                    "       b.AGENTORG,\n" +
                    "       b.AGENT,\n" +
                    "       b.TYPECODE,\n" +
                    "       '' as ipc_desc,\n" +
                    "       if(b.TYPENO='发明专利','发明公布',b.TYPENO),\n" +
                    "       c.LAWSTATUS,\n" +
                    "       c.LAWSTATUSDATE,\n" +
                    "       '' as primary_examiner,\n" +
                    "       '' as assiant_examiner,\n" +
                    "       '' as patent_image,\n" +
                    "       '' as other_references,\n" +
                    "       b.SUMMARY,\n" +
                    "       '' as patent_legal_history,\n" +
                    "       '' as document_types,\n" +
                    "       '' as cites,\n" +
                    "       b.JOBID,\n" +
                    "       \"1\" as isadd\n" +
                    "FROM\n" +
                    "  (SELECT *\n" +
                    "   FROM a_dc_ep_ods.zs_t_sipo_patentinfo\n" +
                    "   WHERE CREDITCODE IS NOT NULL\n" +
                    "     AND CREDITCODE !='') b\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.zs_t_sipo_patent_lawstateinfo c on b.REG_NO = c.REG_NO\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.t_eci_company d on upper(trim(d.credit_code)) = upper(trim(b.CREDITCODE))\n" +
                    "LEFT JOIN\n" +
                    "  (SELECT m.*,n.credit_code\n" +
                    "   FROM a_dc_ep_dwi.t_patent m\n" +
                    "INNER JOIN\n" +
                    "   a_dc_ep_ods.t_eci_company n\n" +
                    "ON upper(trim(m.company_id)) = upper(trim(n.company_id))\n" +
                    "   WHERE n.credit_code IS NOT NULL\n" +
                    "     AND n.credit_code!='') a ON upper(trim(a.credit_code)) = upper(trim(b.creditcode)) and a.application_number = b.REG_NO and a.publication_number = b.PUBCODE\n" +
                    "WHERE a.credit_code IS NULL AND d.credit_code IS NOT NULL and d.credit_code!=''");
            insert_data.createOrReplaceTempView("tmp_t_patent1");
            spark.sql("insert into table a_dc_ep_incr.t_patent select * from tmp_t_patent1");
            //获取中数删除数据表中的数据
            Dataset<Row> delete_data = spark.sql("select a.id,a.key_no,a.company_id,a.company_name,a.patent_id,a.title,a.application_number,a.application_date,a.publication_number,a.publication_date,a.inventor_string_list,a.assigneestring_list,a.agency,a.agent,a.ipc_list,a.ipc_desc,a.kind_code_desc,a.legal_status_desc,a.legal_status_date,a.primary_examiner,a.assiant_examiner,a.patent_image,a.other_references,a.abstracts,a.patent_legal_history,a.document_types,a.cites,a.dates,'-1' as isadd from (select * from a_dc_ep_dwi.t_patent)a inner join (select m.* from a_dc_ep_ods.zs_t_sipo_patentinfo_del m left join a_dc_ep_ods.zs_t_sipo_patentinfo n on m.REG_NO = n.REG_NO where n.REG_NO is null)b on upper(trim(a.application_number)) = upper(trim(b.REG_NO))");
            delete_data.createOrReplaceTempView("tmp_t_patent2");
            spark.sql("insert into table a_dc_ep_incr.t_patent select * from tmp_t_patent2");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_patent");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }

        spark.stop();
    }
}
